package com.song.stormdemo1;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.utils.Utils;

/**
 * Hello world!
 *
 */
public class StormDemo1 
{
    public static void main( String[] args )
    {
    	String DATASOURCE_SPOUT = DataSourceSpout.class.getSimpleName();  
        String SPLIT_BOLD = SplitBolt.class.getSimpleName();  
        String COUNT_BOLT = CountBolt.class.getSimpleName();  
          
        final TopologyBuilder builder = new TopologyBuilder();  
        builder.setSpout(DATASOURCE_SPOUT, new DataSourceSpout());  
        builder.setBolt(SPLIT_BOLD, new SplitBolt()).shuffleGrouping(DATASOURCE_SPOUT);  
        builder.setBolt(COUNT_BOLT, new CountBolt()).shuffleGrouping(SPLIT_BOLD);  
          
        final LocalCluster localCluster = new LocalCluster();  
        final Config config = new Config();  
        localCluster.submitTopology(StormDemo1.class.getSimpleName(), config, builder.createTopology());  
        Utils.sleep(9999999);  
        localCluster.shutdown();  
    }
}
